package com.zzw.controller;

import com.zzw.bean.MqttMessage;
import com.zzw.utils.MqttUtil;
import com.zzw.utils.RedisUtil;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.*;

import javax.servlet.http.HttpServletRequest;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * MQTT控制器
 */
@Controller
@RequestMapping("/mqtt")
public class MqttController {
    private static final Logger logger = LoggerFactory.getLogger(MqttController.class);

    // 每个主题最多存储的消息数量
    private static final int MAX_MESSAGES_PER_TOPIC = 100;

    @Autowired
    private MqttUtil mqttUtil;

    // 存储接收到的消息
    private final ConcurrentHashMap<String, List<MqttMessage>> receivedMessages = new ConcurrentHashMap<>();
    // 存储已订阅的主题
    private final Set<String> subscribedTopics = Collections.synchronizedSet(new HashSet<>());

    /**
     * MQTT管理页面
     */
    @GetMapping("/list")
    public String list(HttpServletRequest request) {
        request.setAttribute("isConnected", mqttUtil.isConnected());
        request.setAttribute("subscribedTopics", subscribedTopics);
        return "mqtt/list";
    }

    /**
     * 发布消息
     */
    @PostMapping("/publish")
    @ResponseBody
    public Map<String, Object> publish(@RequestBody MqttMessage message) {
        Map<String, Object> result = new HashMap<>();

        try {
            // 设置消息ID和发送时间
            message.setId(UUID.randomUUID().toString());
            message.setSendTime(new Date());

            // 发布消息
            boolean success = mqttUtil.publish(message.getTopic(), message.getPayload(), message.getQos(), message.isRetained());

            if (success) {
                message.setStatus(1); // 已发送
                result.put("success", true);
                result.put("message", "消息发送成功");
            } else {
                message.setStatus(2); // 发送失败
                result.put("success", false);
                result.put("message", "消息发送失败");
            }

            result.put("data", message);
        } catch (Exception e) {
            logger.error("发布消息异常: {}", e.getMessage(), e);
            result.put("success", false);
            result.put("message", "发布消息异常: " + e.getMessage());
        }

        return result;
    }

    /**
     * 订阅主题
     */
    @PostMapping("/subscribe")
    @ResponseBody
    public Map<String, Object> subscribe(@RequestParam String topic, @RequestParam int qos) {
        Map<String, Object> result = new HashMap<>();

        try {
            // 检查是否已订阅
            if (subscribedTopics.contains(topic)) {
                result.put("success", false);
                result.put("message", "已订阅该主题");
                return result;
            }

            // 创建消息监听器
            IMqttMessageListener listener = (t, message) -> {
                String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
                logger.info("收到消息: 主题={}, 内容={}", t, payload);

                // 创建消息对象
                MqttMessage mqttMessage = new MqttMessage();
                mqttMessage.setId(UUID.randomUUID().toString());
                mqttMessage.setTopic(t);
                mqttMessage.setPayload(payload);
                mqttMessage.setQos(message.getQos());
                mqttMessage.setRetained(message.isRetained());
                mqttMessage.setSendTime(new Date());
                mqttMessage.setStatus(1);

                // 存储消息
                List<MqttMessage> messages = receivedMessages.computeIfAbsent(t,
                        k -> Collections.synchronizedList(new ArrayList<>()));

                // 添加新消息
                messages.add(mqttMessage);

                // 检查消息数量是否超过上限
                if (messages.size() > MAX_MESSAGES_PER_TOPIC) {
                    // 按时间排序
                    messages.sort(Comparator.comparing(MqttMessage::getSendTime));
                    // 删除最旧的消息
                    messages.remove(0);
                    logger.debug("主题 {} 的消息数量超过上限，已删除最旧的消息", t);
                }
            };

            // 订阅主题
            boolean success = mqttUtil.subscribe(topic, qos, listener);

            if (success) {
                subscribedTopics.add(topic);
                result.put("success", true);
                result.put("message", "订阅成功");
            } else {
                result.put("success", false);
                result.put("message", "订阅失败");
            }
        } catch (Exception e) {
            logger.error("订阅主题异常: {}", e.getMessage(), e);
            result.put("success", false);
            result.put("message", "订阅主题异常: " + e.getMessage());
        }

        return result;
    }

    /**
     * 取消订阅
     */
    @PostMapping("/unsubscribe")
    @ResponseBody
    public Map<String, Object> unsubscribe(@RequestParam String topic) {
        Map<String, Object> result = new HashMap<>();

        try {
            // 检查是否已订阅
            if (!subscribedTopics.contains(topic)) {
                result.put("success", false);
                result.put("message", "未订阅该主题");
                return result;
            }

            // 取消订阅
            boolean success = mqttUtil.unsubscribe(topic);

            if (success) {
                subscribedTopics.remove(topic);
                result.put("success", true);
                result.put("message", "取消订阅成功");
            } else {
                result.put("success", false);
                result.put("message", "取消订阅失败");
            }
        } catch (Exception e) {
            logger.error("取消订阅异常: {}", e.getMessage(), e);
            result.put("success", false);
            result.put("message", "取消订阅异常: " + e.getMessage());
        }

        return result;
    }

    /**
     * 获取接收到的消息
     */
    @GetMapping("/messages")
    @ResponseBody
    public Map<String, Object> getMessages(@RequestParam(required = false) String topic) {
        Map<String, Object> result = new HashMap<>();

        try {
            List<MqttMessage> messages;

            if (topic != null && !topic.isEmpty()) {
                // 获取指定主题的消息
                messages = receivedMessages.getOrDefault(topic, Collections.emptyList());
            } else {
                // 获取所有主题的消息
                messages = new ArrayList<>();
                for (List<MqttMessage> topicMessages : receivedMessages.values()) {
                    messages.addAll(topicMessages);
                }
                // 按时间排序
                messages.sort(Comparator.comparing(MqttMessage::getSendTime).reversed());
            }

            result.put("success", true);
            result.put("data", messages);
        } catch (Exception e) {
            logger.error("获取消息异常: {}", e.getMessage(), e);
            result.put("success", false);
            result.put("message", "获取消息异常: " + e.getMessage());
        }

        return result;
    }

    /**
     * 清空消息
     */
    @PostMapping("/clear")
    @ResponseBody
    public Map<String, Object> clearMessages(@RequestParam(required = false) String topic) {
        Map<String, Object> result = new HashMap<>();

        try {
            if (topic != null && !topic.isEmpty()) {
                // 清空指定主题的消息
                receivedMessages.remove(topic);
            } else {
                // 清空所有消息
                receivedMessages.clear();
            }

            result.put("success", true);
            result.put("message", "清空消息成功");
        } catch (Exception e) {
            logger.error("清空消息异常: {}", e.getMessage(), e);
            result.put("success", false);
            result.put("message", "清空消息异常: " + e.getMessage());
        }

        return result;
    }

    /**
     * 检查连接状态
     */
    @GetMapping("/status")
    @ResponseBody
    public Map<String, Object> checkStatus() {
        Map<String, Object> result = new HashMap<>();
        result.put("connected", mqttUtil.isConnected());
        return result;
    }

    /**
     * 重新连接
     */
    @PostMapping("/reconnect")
    @ResponseBody
    public Map<String, Object> reconnect() {
        Map<String, Object> result = new HashMap<>();

        try {
            boolean success = mqttUtil.reconnect();

            if (success) {
                result.put("success", true);
                result.put("message", "重新连接成功");
            } else {
                result.put("success", false);
                result.put("message", "重新连接失败");
            }
        } catch (Exception e) {
            logger.error("重新连接异常: {}", e.getMessage(), e);
            result.put("success", false);
            result.put("message", "重新连接异常: " + e.getMessage());
        }

        return result;
    }
}
